import json import time import datetime import requests from collections import Counter from gm.api import * now = time.time() print(f"kpl_api开始运行--{now}") # 竞价 DABAN_TYPE_BIDDING = 8 # 涨停 DABAN_TYPE_LIMIT_UP = 1 # 炸板 DABAN_TYPE_OPEN_LIMIT_UP = 2 # 跌停 DABAN_TYPE_LIMIT_DOWN = 3 # 曾跌停 DABAN_TYPE_EVER_LIMIT_DOWN = 5 def __base_request(url, data): headers = { "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "User-Agent": "Dalvik / 2.1.0(Linux;U;Android 6.0.1;MuMu Build/V417IR)" } # proxies={'https': '192.168.3.251:9002'} # 禁止代理，不然会走本地代理 response = requests.post(url, data=data, headers=headers, proxies={"http": None, "https": None}) if response.status_code != 200: raise Exception("请求出错") return response.text def daBanList(pidType): data = "Order=1&a=DaBanList&st=100&c=HomeDingPan&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23" \ f"&VerSion=5.8.0.2&Index=0&Is_st=1&PidType={pidType}&apiv=w32&Type=4&FilterMotherboard=0&Filter=0&FilterTIB=0" \ "&FilterGem=0 " result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 市场行情-行业 主力净额由高到低排序 实时获取 def getMarketIndustryRealRankingInfo(orderJingE_DESC=True): data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=20&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=4&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 市场行情-精选 主力净额由高到低排序 实时获取 st=20 请求地址中 st=20 代表获取前20（排序方式为主力净额降序排列） # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=10&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # return result # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=3&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # return result # #市场行情-精选 （排序方式为强度降序排列） st=20 请求地址中 st=20 代表获取前20 def getMarketJingXuanRealRankingInfo(): data = f"Order=1&a=RealRankingInfo&st=10&a=RealRankingInfo&apiv=w35&Type=1&Index=0&c=ZhiShuRanking&VerSion=5.13.0.0&Order=1&PhoneOSNew=1&ZSType=7&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9773&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 获取代码的板块 def getStockIDPlate(code): data = f"a=GetStockIDPlate_New&apiv=w32&c=StockL2Data&StockID={code}&PhoneOSNew=1&UserID=0&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Token=0&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) if int(result["errcode"]) != 0: return None return result["ListJX"] if result["ListJX"] else result["List"] # IDPlate = getStockIDPlate(600550) # print(f"IDPlate===={IDPlate}") # 获取代码的精选板块 # 返回格式：[(板块代码,板块名称,涨幅百分比)] def getCodeJingXuanBlocks(code): data = f"a=GetStockIDPlate&apiv=w32&Type=2&c=StockL2Data&StockID={code}&PhoneOSNew=1&UserID=0&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Token=0&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) return result.get("ListJX") # 获取该概念下的个股代码及其他 def getCodesByPlate(plate_code): data = f"Order=1&a=ZhiShuStockList_W8&st=30&c=ZhiShuRanking&PhoneOSNew=1&old=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&IsZZ=0&Token=0&Index=0&apiv=w32&Type=6&IsKZZType=0&UserID=0&PlateID={plate_code}&" return __base_request("https://apphq.longhuvip.com/w1/api/index.php",data=data) # 获取概念中的板块中的子板块 def getSonPlate(plate_code): data = f"a=SonPlate_Info&apiv=w32&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&PlateID={plate_code}&" return __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) # 市场行情-行业 # def getMarketIndustryRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=80&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=4&" # return __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # # # 市场行情-精选 # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=80&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # return __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # 获取自由流通市值 def getZYLTAmount(code): data = f"a=GetStockPanKou_Narrow&apiv=w32&c=StockL2Data&VerSion=5.8.0.2&State=1&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&StockID={code}&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) if "real" in result: return result["real"].get("actualcirculation_value") return None # 获取涨停列表 def __getLimitUpInfo(pidType, page, pageSize): data = f"Order=0&a=DailyLimitPerformance&st={pageSize}&apiv=w35&Type=4&c=HomeDingPan&PhoneOSNew=1&DeviceID=a38adabb-99ef-3116-8bb9-6d893c846e24&VerSion=5.13.0.0&Index={(page - 1) * pageSize}&PidType={pidType}&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 整理涨停列表的数据 def getLimitUpInfoNew(): pids = [(1, "首板"), (2, "2连板"), (3, "3连板"), (4, "4连板"), (5, "")] fresults = [] for pid_info in pids: results = [] for i in range(10): start_time = time.time() result = __getLimitUpInfo(pid_info[0], i + 1, 20) # print("请求用时", time.time() - start_time) result = json.loads(result) datas = result["info"][0] results.extend(datas) day = result["info"][1] if len(datas) < 20: break for r in results: if not r[18] and pid_info[1]: r[18] = pid_info[1] # # 替换掉板块名称 机器人概念/机器人 # for i in range(len(r)): # if type(r[i]) == str: # r[i] = kpl_util.filter_block(r[i]) fresults.extend(results) return json.dumps({"errcode": 0, "list": fresults}) # if __name__ == "__main__": # print(f"打板列表t(pidType)====={daBanList(2)}") # print(f"获取个股代码的板块==={getStockIDPlate('002766')}") # print((f"获取个股代码的精选板块==={getCodeJingXuanBlocks('002878')}")) # print(f"获取该概念下的个股代码及其他====={getCodesByPlate(885500)}") 《《《《《《《《《《 # print(f"获取概念中的板块中的子板块====={json.loads(getSonPlate(801085))}") # print(f"获取概念中的板块强度====={getSonPlate(getCodesByPlate(getCodeJingXuanBlocks('002452')[2][0]))}") # print(f"市场行情-行业板块 数==={len(getMarketIndustryRealRankingInfo(True))}") # print(f"市场行情-行业板块==={json.loads(getMarketIndustryRealRankingInfo(True))}") # 返回格式：['板块ID','板块名称','强度','涨幅','未知','成交额','''''''''强度','未知'] # print(f"市场行情-精选板块 数==={getMarketJingXuanRealRankingInfo(True)}") # print(f"市场行情-精选板块==={json.loads(getMarketJingXuanRealRankingInfo(True))}") # print(f"股票代码：{Market_situation_selected_sectors_No1[0]}") # jingxuanbankuai = json.loads(getMarketJingXuanRealRankingInfo(True)) # print(f"jingxuanbankuai==={type(jingxuanbankuai)}") # print(f"板块代码：{jingxuanbankuai['list'][0][0]}，板块名称：{jingxuanbankuai['list'][0][1]}，强度：{jingxuanbankuai['list'][0][2]}，涨幅：{jingxuanbankuai['list'][0][3]}，未知：{jingxuanbankuai['list'][0][4]}，成交额：{round(jingxuanbankuai['list'][0][5]/100000000)}亿，主力净额：{round(jingxuanbankuai['list'][0][6]/100000000,2)}亿，主买：{round(jingxuanbankuai['list'][0][7]/100000000,2)}亿，主卖：{round(jingxuanbankuai['list'][0][8]/100000000,2)}亿，未知：{jingxuanbankuai['list'][0][9]}，流通值：{round(jingxuanbankuai['list'][0][10]/100000000,2)}亿，未知/或为最大涨跌幅：{round(jingxuanbankuai['list'][0][11],2)}，未知：{round(jingxuanbankuai['list'][0][12]/100000000,2)}亿，总市值：{round(jingxuanbankuai['list'][0][13]/100000000,2)}亿，第一季度机构持仓：{round(jingxuanbankuai['list'][0][14]/100000000,2)}亿，未知：{round(jingxuanbankuai['list'][0][15],2)}，未知：{round(jingxuanbankuai['list'][0][16],2)}，强度：{round(jingxuanbankuai['list'][0][17],2)}") # # 部分板块没有子板块 # print(f"获取概念中的板块中的子板块====={json.loads(getSonPlate(801248))}") # print(f"自由流通市值==={getZYLTAmount('603319')}") # print((f"获取个股代码的精选板块列表==={getCodeJingXuanBlocks('002452')}")) # print((f"获取个股代码的精选第一板块代码==={getCodeJingXuanBlocks('002452')[0][0]}")) # print(f"获取该概念下的个股代码及其他====={json.loads(getCodesByPlate(getCodeJingXuanBlocks('002452')[0][0]))}") # print(f"获取该概念下的个股代码及其他dddddd====={json.loads(its_strongest_sector_situation)}") # print(f"涨停列表及概念板块={json.loads(getLimitUpInfoNew())['list']}") ######################################################################################################################################################################################################################## # 获取行情精选板块 强度排名 def get_market_sift_plate_its_stock_power(): data = (getMarketJingXuanRealRankingInfo()) market_sift_plate = json.loads(data) # print(f"market_sift_plate 数 ======{len(market_sift_plate['list'])}") # print(f"market_sift_plate ======{market_sift_plate['list'][0]}") # 初始化精选板块对应个股字典 market_sift_plate_stock_dict = {} for i in market_sift_plate['list']: plate_name = i[1] # 尝试过滤掉无意义的概念板块 if plate_name != '科创板' and plate_name != '北交所' and plate_name != '次新股' and plate_name != '无' and plate_name != 'ST板块' and plate_name != 'ST摘帽' and plate_name != '并购重组' and plate_name != '国企改革' and plate_name != '超跌' and plate_name != '壳资源' and plate_name != '股权转让' and plate_name != '送转填权' and plate_name.find("增长") > 0: # print(f"{i[1]} 强度:{i[2]}") # 通过板块ID获取其下面的个股强度列表 its_stock = getCodesByPlate(i[0]) its_stock = json.loads(its_stock) # print(f"its_stock['list'] === {its_stock['list']}") print(f"its_stock['list'][0] === {its_stock['list'][0]}") # 获取该概念板块的的子版块 its_son_plate = json.loads(getSonPlate(i[0])) # print(f"its_son_plate=={its_son_plate['List']}") # if its_son_plate['List'][0][2] > 0: # print(f"its_son_plate=={its_son_plate['List'][0]}") # for son in its_son_plate['List']: # if son[2] > 0: # print(f"son===={son}") # 初始化股票强度列表 stock_power_list = [] for s in its_stock['list']: # 过滤掉涨幅大于6.5%且小于0%的 和 名称中包含ST的 和 涨速小于等于0%的 和 只要昨日未涨停 和 上证或深证的正股 if s[6] > 0 and s[6] < 6.5 and s[1].find("ST") < 0 and s[1].find("XD") < 0 and s[9] > 1 and s[23].find("板") < 0 and s[24].find("板") < 0 and (s[0].startswith('60') or s[0].startswith('00')): # print(f"{s[1]}，个股代码：{s[0]}， 涨幅：{s[6]}% 涨速：{s[9]}% 概念：{s[4]} 主力资金推测：{s[2]} 领涨次数：{s[40]} 今日第几板：{s[23]} 是否破版{s[24]}") its_stock_power = [] #对个股强度属性列表惊醒初始化 # 逐个选择性添加its_stock中的元素到个股强度列表中 its_stock_power.append(s[1]) its_stock_power.append(s[0]) its_stock_power.append(s[6]) its_stock_power.append(s[9]) its_stock_power.append(s[4]) its_stock_power.append(s[2]) its_stock_power.append(s[40]) # print(f"its_stock_power===={its_stock_power}") # 整体将添加完善的个股强度列表添加到股票列表中 stock_power_list.append(its_stock_power) # print(f"its_stock_power_list===={its_stock_power_list}") # 过滤掉没有瞬时高强度个股的空概念 if len(stock_power_list) != 0: # 将对应板块的股票强度列表新建一个字典 stock_power_item = {i[1] : stock_power_list} # 并更新到精选板块个股字典中 market_sift_plate_stock_dict.update(stock_power_item) print(f"精选板块股票强度数据更新 == {market_sift_plate_stock_dict}") return market_sift_plate_stock_dict get_market_sift_plate_its_stock_power() def get_market_sift_plate_its_stock_power_process(callback): while True: try: now = time.time() # print(f"kpl_limit_up_process开始了{now}") its_stock_power = get_market_sift_plate_its_stock_power() callback(its_stock_power) # print(f"精选板块拉升个股更新===={its_stock_power}") except Exception as e: print(f"An error occurred: {e}") finally: time.sleep(3) # 获取涨停信息列表并存储本地的函数 def get_limit_up_info(): # 获取涨停信息列表 limit_up_info = json.loads(getLimitUpInfoNew())['list'] # 提取涨停列表中的板块名称 limit_up_block_names = [] # # 打印数据获取时间，并放到整体列表中 # now = time.time() # limit_up_block_names.append(now) # 循环添加涨停概念 for i in limit_up_info: limit_up_block_names.append(i[5]) # print(f"limit_up_block_names==={limit_up_block_names}") # return limit_up_block_names # # 使用Counter计算每个元素的出现次数 # counter = Counter(limit_up_block_names) # # 找出出现次数最多的元素及其次数 # most_common_element, most_common_count = counter.most_common(1)[0] # # 打印出现次数最多的元素 # print(f"主线概念：{most_common_element}，出现了 {most_common_count} 次") return limit_up_block_names # 将获取到的涨停概念板块转JSON格式并存储在本地文件夹中 # 将字典转换为JSON格式的字符串 # json_data = json.dumps(limit_up_block_names) # if len(json_data) != 0: # # 写入文件 # with open('local_storage_data/limit_up_block_names.json', 'w', encoding='utf-8') as f: # f.write(json_data) # return limit_up_block_names # 为开盘啦接口获取的涨停列表概念板块单独开一个进程 形参(callback) def kpl_limit_up_process(callback): while True: try: now = time.time() # print(f"kpl_limit_up_process开始了{now}") limit_up_block_names = get_limit_up_info() callback(limit_up_block_names) # print(f"涨停更新===={limit_up_block_names}") # print(f"涨停更新数量===={len(limit_up_block_names)}") # print(f"kpl_limit_up_process完成一下{now}") except Exception as e: print(f"An error occurred: {e}") finally: time.sleep(3) # kpl_limit_up_process() # 获取全部个股的板块并存储的函数 def all_stocks_plate_dict(stocks_list): all_stocks_plate_dict = {} # 逐个获取个股精选板块概念和自由市值等，并整体放入一个新创建的字典中然后添加到数据中 for i in stocks_list: try: code = i.split('.')[1] # print(f"i==={i}") # 获取个股的自由市值 free_market_value = getZYLTAmount(code) # 获取个股的板块列表 selected_blocks = getStockIDPlate(code) # 提取精选板块中的板块名称 selected_plate_list = [block[1] for block in selected_blocks] # print(f"selected_block_names==={selected_block_list}") block_data = { 'free_market_value': free_market_value, #添加自由市值 'plate': selected_plate_list #添加精选板块 } # 将code作为键，stocks_selected_block_data作为值添加到stocks_block_data字典中 all_stocks_plate_dict[code] = block_data # print(f"all_stocks_plate_dict==={all_stocks_plate_dict}") except : pass # return stocks_plate_data # print(f"all_stocks_plate_dict==={len(all_stocks_plate_dict)}") # 将获取到的范围票概念板块转JSON格式并存储在本地文件夹中 # 将字典转换为JSON格式的字符串 json_data = json.dumps(all_stocks_plate_dict) # 写入文件 with open('local_storage_data/all_stocks_plate_dict.json', 'w', encoding='utf-8') as f: f.write(json_data) now = datetime.datetime.now() # 获取本机时间 print(f"写入所有个股板块文件完成！::{now}") # 获取全部个股的精选板块并存储的函数 def stocks_list_selected_blocks(min_stocks): stocks_selected_block_data = [] # 逐个获取个股精选板块概念和自由市值等，并整体放入一个新创建的字典中然后添加到数据中 for i in min_stocks: try: code = i.split('.')[1] # 获取个股的自由市值 free_market_value = getZYLTAmount(code) # 获取个股的精选板块列表 # selected_blocks = getCodeJingXuanBlocks('000021') selected_blocks = getCodeJingXuanBlocks(code) # 提取精选板块中的板块名称 selected_block_list = [block[1] for block in selected_blocks] # print(f"selected_block_names==={selected_block_list}") stocks_selected_block_dict = { 'code': code, #添加股票代码 'free_market_value': free_market_value, #添加自由市值 'selected_block': selected_block_list #添加精选板块 } stocks_selected_block_data.append(stocks_selected_block_dict) # print(f"stocks_selected_block_data==={stocks_selected_block_dict}") except : pass # print(f"stocks_selected_block_data==={len(stocks_selected_block_data)}") # 将获取到的范围票概念板块转JSON格式并存储在本地文件夹中 # 将字典转换为JSON格式的字符串 json_data = json.dumps(stocks_selected_block_data) # 写入文件 with open('local_storage_data/stocks_selected_block_data.json', 'w', encoding='utf-8') as f: f.write(json_data) now = datetime.datetime.now() # 获取本机时间 print(f"写入精选板块文件完成！::{now}") # kpl_stocks_list_selected_blocks_process() #在 kpl_api.py中可以调用 # stocks_list_selected_blocks(min_stocks) #在 kpl_api.py中可以调用 # list = ['SHSE.600805','SHSE.600804'] # # all_stocks_plate_dict(list)